import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object zk1 {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("app")
    val sc = new SparkContext(sparkConf)
    sc.setLogLevel("ERROR")


    var lineText:RDD[String]=sc.textFile("data/traffic-data.txt")
    //3.	数据加载：统计交通数据条目数和显示前10天样本数据
    println(lineText.count)

    var a:Array[String]=lineText.take(10)
    for (elem<-a){
      println(elem)
    }

    var mapRDD: RDD[traffic] = lineText.filter(o => {
      o.split(",").length == 7
    }).map(o => {
      var aa = o.split(",")
      traffic(aa(0).toInt, aa(1).toInt, aa(2), aa(3), aa(4).toInt, aa(5).toInt, aa(6).toInt)
    })

    //mapRDD.foreach(println)

    //4.	分析1：基于Scala代码编写，调用RDD算子，获取车速大于90以上交通数据
    var rdd1:RDD[traffic]=mapRDD.filter(o=>o.speed>90)
    rdd1.foreach(println)

//5.	分析2：基于Scala代码编写，调用RDD算子，计算各个区域交通流量
  var rddGroupby: RDD[(String, Iterable[traffic])] =
    rdd1.groupBy(a => a.quid.toString)

    var rddGroupbyValues: RDD[(String, Int)] = rddGroupby.map(a => {
      (a._1, a._2.size)
    })
    rddGroupbyValues.foreach(println)

    //6.	分析2：对第5题计算结果，按照交通流量降序排序，获取Top3流量区域
    val rdd2:Array[(String,Int)] = rddGroupbyValues.sortBy(o => (o._2), false).take(3)
    rdd2.foreach(println)

    //7.	分析3：基于Scala代码编写，调用RDD算子，计算各个省份车牌的交通流量
    var rddGroupby2: RDD[(String, Iterable[traffic])] =
      rdd1.groupBy(a => a.cp.split("-")(0))

    var rddGroupbyValues2: RDD[(String, Int)] = rddGroupby2.map(a => {
      (a._1, a._2.size)
    })
    rddGroupbyValues2.foreach(println)
    //8.	分析3：对第7题计算结果，保存HDFS文件系统，并查看文件数据

    //rddGroupbyValues2.saveAsTextFile("hdfs://cdh01:8020/tmp/data2")






    sc.stop()
  }

}
case class traffic(jcid:Int,jkid:Int,cp:String,zp_time:String,speed:Int,luid:Int,quid:Int)